chore: event count throttle for squashed commands #4924

kostasrim · 2025-04-11T14:48:42Z

Throttle/preempt flows that use multi command squasher and crb crosses the limit.

Signed-off-by: kostas <[email protected]>

kostasrim · 2025-04-22T11:39:06Z

src/server/multi_command_squasher.cc

+thread_local size_t MultiCommandSquasher::throttle_size_limit_ =
+    absl::GetFlag(FLAGS_throttle_squashed);
+
+thread_local util::fb2::EventCount MultiCommandSquasher::ec_;

 MultiCommandSquasher::MultiCommandSquasher(absl::Span<StoredCmd> cmds, ConnectionContext* cntx,


This is used not only from async fiber but also directly from the connection. If we preempt, the connection will also "freeze". I guess this is fine, just mentioning it here for completeness.

There are 3 calls of this and all of them should be ok if we preempt from these flows.

kostasrim · 2025-04-22T11:39:19Z

tests/dragonfly/memory_test.py

+        await cl.execute_command("exec")
+
+    # With the current approach this will overshoot
+    #    await client.execute_command("multi")


I wish we also handled this case as well

what is the difference? why does it not handles this case?

kostasrim · 2025-04-22T11:39:50Z

src/server/multi_command_squasher.h

@@ -94,6 +104,9 @@ class MultiCommandSquasher {

  // we increase size in one thread and decrease in another
  static atomic_uint64_t current_reply_size_;
+  static thread_local size_t throttle_size_limit_;
+  // Used to throttle when memory is tight
+  static thread_local util::fb2::EventCount ec_;


We need this to avoid ThisFiber::Yield, ThisFiber::SleepFor in while(true).

since it's thread local, it's more efficient to use NoOpLock together with CondVarAny

@romange nice!

Actually, I added a bug here:

static atomic_uint64_t current_reply_size_;

so current_reply_size is not thread local. So what can happen is:

Core 0 -> starts multi/exec
Core 1 -> starts multi/exec but needs to throttle so it goes to sleep waiting on the thread local cond variable
Core 0 -> is done, notifies the thread local
Core 1 -> the fiber never awakes even though we decremented current_reply_size.

Since current_reply_size is global then so should ec_.

P.s. not very happy with this extra synchronization but we only pay it when we are under memory pressure

it should be thread local.

kostasrim · 2025-04-22T11:40:24Z

@adiholden pinging for an early discussion here

kostasrim · 2025-04-22T11:41:42Z

src/server/multi_command_squasher.cc

@@ -15,6 +16,8 @@
 #include "server/transaction.h"
 #include "server/tx_base.h"

+ABSL_FLAG(size_t, throttle_squashed, 0, "");


@adiholden I will adjust as we said f2f. Looking for some early feedback based on our discussion

adiholden · 2025-04-22T12:01:41Z

src/server/multi_command_squasher.cc

@@ -63,6 +66,10 @@ size_t Size(const facade::CapturingReplyBuilder::Payload& payload) {
 }  // namespace

 atomic_uint64_t MultiCommandSquasher::current_reply_size_ = 0;
+thread_local size_t MultiCommandSquasher::throttle_size_limit_ =
+    absl::GetFlag(FLAGS_throttle_squashed);


As discussed this morning multiply by thread count. The limit should be per thread and the current_reply_size_ is global counter

Yes I know, I even wrote a comment above that I will follow up with this 😄

I wanted to know if you have anything else to add 😄

adiholden · 2025-05-04T06:10:59Z

src/server/multi_command_squasher.cc

@@ -63,6 +66,9 @@ size_t Size(const facade::CapturingReplyBuilder::Payload& payload) {
 }  // namespace

 atomic_uint64_t MultiCommandSquasher::current_reply_size_ = 0;
+thread_local size_t MultiCommandSquasher::throttle_size_limit_ =


I believe we should multiply throttle_squashed by the number of io threads and not shard number

we should no do this at all. the limit should be by thread. and in general it's not well defined to initialize thread local by using another thread local that is initialized to nullptr.

adiholden · 2025-05-04T06:56:25Z

src/server/multi_command_squasher.h

@@ -37,6 +38,15 @@ class MultiCommandSquasher {
    return current_reply_size_.load(std::memory_order_relaxed);
  }

+  static bool IsMultiCommandSquasherOverLimit() {


maybe rename to IsReplySizeOverLimit?

adiholden · 2025-05-04T06:58:03Z

src/server/multi_command_squasher.h

+  // Used to throttle when memory is tight
+  static util::fb2::EventCount ec_;
+
+  static thread_local size_t throttle_size_limit_;


maybe reply_size_limit_ ?

adiholden · 2025-05-04T07:01:45Z

src/server/multi_command_squasher.cc

@@ -15,6 +16,8 @@
 #include "server/transaction.h"
 #include "server/tx_base.h"

+ABSL_FLAG(size_t, throttle_squashed, 0, "");


maybe squashed_reply_size_limit
add flag description
also I think we should have a default limit here maybe 128_MB ?

adiholden · 2025-05-04T07:29:38Z

tests/dragonfly/memory_test.py

+        # At any point we should not cross this limit
+        assert df.rss < 1_500_000_000
+        cl = df.client()
+        await cl.execute_command("multi")


I see that the flow that you are testing is the multi exec flow which I did not think about. When I suggested this throttling I was thinking about the pipeline flow.
When reviewing now the multi exec flow I am not 100% sure for implying this logic in this flow as we when you do the await in the code to wait for the size to decrease we already scheduled the transaction and I am not sure if this can lead in some cases to a deadlock

romange · 2025-05-04T09:12:10Z

src/server/server_state.h

@@ -270,6 +270,10 @@ class ServerState {  // public struct - to allow initialization.

  bool ShouldLogSlowCmd(unsigned latency_usec) const;

+  size_t GetTotalShards() const {


not needed - you can use shard_set->size() everywhere

romange · 2025-05-04T09:19:16Z

src/server/multi_command_squasher.cc

@@ -215,6 +222,9 @@ bool MultiCommandSquasher::ExecuteSquashed(facade::RedisReplyBuilder* rb) {
  if (order_.empty())
    return true;

+  MultiCommandSquasher::ec_.await(


all our current approaches of limiting memory are "per-thread", this is consistent and works nicely with shared-nothing. What is the reason for not using per thread limits? In addition, we already have per thread throttling inside dragonfly_connection code, see IsPipelineBufferOverLimit. Did you consider pigging back on this mechanism instead ?

as long as current_reply_size_ is global I dont see how we can do this per thread

romange · 2025-05-04T09:23:04Z

src/server/multi_command_squasher.cc

  for (auto idx : order_) {
    auto& replies = sharded_[idx].replies;
    CHECK(!replies.empty());

    aborted |= opts_.error_abort && CapturingReplyBuilder::TryExtractError(replies.back());

-    current_reply_size_.fetch_sub(Size(replies.back()), std::memory_order_relaxed);


I said nothing when current_reply_size_ was added. it was a mistake. I do not want anyone introduces global states in Dragonfly codebase.

@BorysTheDev FYI

but @romange we discussed this when current_reply_size_ was added. Because the multi command sqasher is adding replies in different threads you said it makes sense.

Would something like this work?
https://github.com/dragonflydb/dragonfly/compare/RemoveAtomic?expand=1

@romange The change in your branch linked above is that you count the reply size after we executed all the squashed commands. So it does uses the thread local approach correctly but what we expose to metrics is not accurate because at the time the capture reply builders grow we do not expose this until we finish with all the squashed. I think that applying such logic will impact the throttling on reply size just in delay, we will throttle but not when actually we are at the threshold but with some delay. I guess we can do this change.

There is no a "correct" solution because I can show you a scenario where single central atomic won't work either:
we throttle before we send commands to shards, but maybe there are tons of squashed commands in flight that have not filled their replies yet, so you let the next command pass and only then the reply buffer increases.

I would rather have a less accurate metric than have have all our threads contend on atomics and now on a single condvar. this kills performance. I won't be surprised that even now squashing performance is worse because of the "reply bufffer size" atomic being hammered by multiple threads.

kostasrim self-assigned this Apr 11, 2025

kostasrim force-pushed the kpr1 branch from 56acf77 to 7f19ba2 Compare April 22, 2025 11:37

kostasrim added 2 commits April 22, 2025 14:37

chore: event count throttle for squashed commands

8f53c94

Signed-off-by: kostas <[email protected]>

Merge branch 'main' into kpr1

8aa26b2

kostasrim force-pushed the kpr1 branch from 7f19ba2 to 8aa26b2 Compare April 22, 2025 11:37

kostasrim commented Apr 22, 2025

View reviewed changes

kostasrim changed the title ~~[experiment do not review] chore: reject squashed when crb exceeds limit~~ chore: reject squashed when crb exceeds limit Apr 22, 2025

kostasrim marked this pull request as ready for review April 22, 2025 11:40

kostasrim requested a review from adiholden April 22, 2025 11:40

kostasrim changed the title ~~chore: reject squashed when crb exceeds limit~~ chore: event count throttle for squashed commands Apr 22, 2025

kostasrim commented Apr 22, 2025

View reviewed changes

adiholden reviewed Apr 22, 2025

View reviewed changes

fixes

2faa2a0

kostasrim requested review from romange and adiholden April 30, 2025 11:42

adiholden reviewed May 4, 2025

View reviewed changes

romange reviewed May 4, 2025

View reviewed changes

kostasrim added 2 commits May 6, 2025 14:01

Merge branch 'main' into kpr1

8e3134e

comments

44c607a

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

chore: event count throttle for squashed commands #4924

chore: event count throttle for squashed commands #4924

kostasrim commented Apr 11, 2025 •

edited

Loading

kostasrim Apr 22, 2025

kostasrim Apr 22, 2025

adiholden May 4, 2025

kostasrim Apr 22, 2025

romange Apr 22, 2025

kostasrim Apr 22, 2025

kostasrim Apr 23, 2025

romange May 4, 2025

kostasrim commented Apr 22, 2025

kostasrim Apr 22, 2025

adiholden Apr 22, 2025

kostasrim Apr 22, 2025

adiholden May 4, 2025

romange May 4, 2025

adiholden May 4, 2025

adiholden May 4, 2025

adiholden May 4, 2025

adiholden May 4, 2025

romange May 4, 2025

romange May 4, 2025

adiholden May 4, 2025

romange May 4, 2025

romange May 4, 2025

adiholden May 4, 2025

romange May 4, 2025

adiholden May 5, 2025

romange May 5, 2025

		@@ -270,6 +270,10 @@ class ServerState { // public struct - to allow initialization.

		bool ShouldLogSlowCmd(unsigned latency_usec) const;

		size_t GetTotalShards() const {

chore: event count throttle for squashed commands #4924

Are you sure you want to change the base?

chore: event count throttle for squashed commands #4924

Conversation

kostasrim commented Apr 11, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kostasrim commented Apr 22, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kostasrim commented Apr 11, 2025 •

edited

Loading